/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

import org.slf4j.Logger;
import org.sss.common.io.DBSet;
import org.sss.common.io.KeyFormatter;
import org.sss.mapreduce.DB;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.Encodings;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.Packable;

public class TupleReader<K extends Packable, V extends Packable> implements Iterator<Tuple<K, V>> {
  @SuppressWarnings("unused")
  private static Logger logger = SssServer.getLogger(TupleReader.class);

  private List<byte[]> tuples;
  private int index;
  private Decoder keyDec;
  private Decoder valDec;
  private Class<K> keyClass;
  private Class<V> valueClass;
  private KeyFormatter keyFormat;
  private Updater updater;
  private final int count_max;
  private final int limitNBytes;
  private final boolean destructive;

  public TupleReader(SssServerResource sssResouce,
                  StorageNode nearNode,
                  GroupID gid,
                  Encodings encodings,
                  Class<K> keyClass,
                  Class<V> valueClass) throws SssException {
    List<StorageNode> nodes = gid.isBroadcast()?Arrays.asList(nearNode):sssResouce.getStorageNodeList();
    if (sssResouce.getServerConfiguration().ioConf.db_use_vanilla) {
      updater = new UpdaterVanilla(sssResouce.getDBSetForSideData(), nodes, gid);
    } else {
      updater = new UpdaterNormal(sssResouce.getDBSetForSideData(), nodes, gid);
    }
    this.keyClass = keyClass;
    this.valueClass = valueClass;
    this.keyDec = encodings.createDecoder(keyClass);
    this.valDec = encodings.createDecoder(valueClass);
    this.tuples = null;
    this.index = 0;
    this.count_max = sssResouce.getServerConfiguration().ioConf.get_batchcount;
    this.limitNBytes = sssResouce.getServerConfiguration().ioConf.get_limit_nbytes;
    this.destructive = false;
    this.keyFormat = KeyFormatter.get(gid.getKeyFormat());
  }

  @Override
  public boolean hasNext() {
    while (true){
      if (tuples != null && index < tuples.size()) {
          return true;
      }
      index = 0;
      tuples = updater.update();
      if (tuples == null) {
        return false;
      }
    }
  }

  @Override
  public Tuple<K, V> next() {
    if (!hasNext()) {
      throw new NoSuchElementException();
    }
    assert (index + 1) < tuples.size();

    try {
      Tuple<K, V> t = new Tuple<K, V>(
          keyClass.cast(keyDec.decode(keyFormat.getUserKey(tuples.get(index)))),
          valueClass.cast(valDec.decode(tuples.get(index + 1))));
      index += 2;
      return t;
    }
    catch (IOException e) {
      throw new RuntimeException(e);
    }
  }


  @Override
  public void remove() {
    throw new UnsupportedOperationException();
  }

  private abstract class Updater {
    private DBSet dbSet;
    private DB db;
    private Iterator<StorageNode> nodeIt;

    public Updater(DBSet dbSet, List<StorageNode> nodes) {
      this.dbSet = dbSet;
      this.nodeIt = nodes.iterator();
      this.db = null;
    }

    protected DB getDB() {
      return db;
    }

    protected DB nextDB() {
      try {
        if (!nodeIt.hasNext()) {
          this.db = null;
        } else {
          this.db = dbSet.get(nodeIt.next());
        }
        return db;
      } catch (IOException e) {
        this.db = null;
        throw new RuntimeException(e);
      }
    }

    public abstract List<byte[]> update();
  }

  private class UpdaterNormal extends Updater {
    private byte[] prefix;
    private byte[] sentinel;
    private byte[] cur;

    public UpdaterNormal(DBSet dbSet, List<StorageNode> nodes, GroupID gid) throws SssException {
      super(dbSet, nodes);
      try {
        this.prefix = KeyFormatter.getGroupIDPrefix(gid);
        this.sentinel = KeyFormatter.getGroupIDSentinel(gid);
        this.cur = null;
      } catch (IOException e) {
        throw new SssException(e);
      }
    }

    @Override
    public List<byte[]> update() {
      try {
        List<byte[]> tuples = null;
        DB db = getDB();
        boolean initial = false;
        while (true) {
          if (cur == null) {
            db = nextDB();
            if (db == null) {
              return null;
            }
            initial = true;
            cur = prefix;
          }
          tuples = destructive ? db.ranged(cur, sentinel, limitNBytes, initial)
                               : db.range(cur, sentinel, limitNBytes, initial);
          if (tuples.size() > 0) {
            cur = tuples.get(tuples.size() - 2);
            return tuples;
          } else {
            cur = null;
          }
          initial = false;
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private class UpdaterVanilla extends Updater {
    private byte[] prefix;
    private List<byte[]> allkeys = null;
    private int indexAtDB = 0;

    public UpdaterVanilla(DBSet dbSet, List<StorageNode> nodes, GroupID gid) throws SssException {
      super(dbSet, nodes);
      try {
        this.prefix = KeyFormatter.getGroupIDPrefix(gid);
      } catch (IOException e) {
        throw new SssException(e);
      }
    }

    @Override
    public List<byte[]> update() {
      try {
        List<byte[]> tuples = null;
        DB db = getDB();
        while (true) {
          if (allkeys == null || allkeys.size() == indexAtDB) {
            db = nextDB();
            if (db == null) {
              return null;
            }
            allkeys = db.fwmkeys(prefix, -1);
          }
          int n = Math.min(allkeys.size() - indexAtDB, count_max);
          byte[][] ks = allkeys.subList(indexAtDB, indexAtDB + n).toArray(new byte[n][]);

          tuples = db.getlist(ks);
          this.indexAtDB += n;
          if (destructive) {
            db.outlist(ks);
          }
          return tuples;
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  }
}
